package com.stars.distributed.schedule.client.demo.rocketmq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson.parser.Feature;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.jy.modules.rocketmq.consumer.AbstractRocketMQConsumer;
import com.jy.modules.rocketmq.util.StringUtils;
import com.stars.distributed.schedule.client.DistributedTaskExecutionContent;
import com.stars.distributed.schedule.client.DistributedTaskExecutionResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

@Component
@Configuration
@PropertySource("classpath:rocketMQ.properties")
public class MQDistributedConsumer extends AbstractRocketMQConsumer{
    private static final Logger logger =  LoggerFactory.getLogger(MQDistributedConsumer.class);

    @Value("${com.consumer.distributed.groupName}")
    protected String groupName; // consumerGroupName
    @Value("${rocketmq.nameServer}")
    protected String namesrvAddr; // namesrvAddr
    @Value("${com.consumer.distributed.topic}")
    protected String topics;  // 主题

    @Autowired
    MQDistributedCallBackProducer callBackProducer;

    public MQDistributedConsumer(){
    }
    public MQDistributedConsumer(String groupName, String namesrvAddr, String topics) {
        this.groupName = groupName;
        this.namesrvAddr = namesrvAddr;
        this.topics = topics;
    }
    @Bean(name="creditListener")
    public String container(){
        this.createListenerContainer(groupName, namesrvAddr, topics);
        return "container success";
    }
    /**
     * Message rocketMessage = new Message(topic, JSON.toJSONString(DistributedTaskExecutionContent).getBytes("utf-8"));
     * 实际的业务处理类，发送端的数据 DistributedTaskExecutionContent 实体
     */
    @Override
    public void doExecute(MessageExt msg) {
        long beginTime = new Date().getTime();
        String returnJson;
        logger.info("{} ,topics: {}, tags: {},  beginTime: {}" , Thread.currentThread().getName() , msg.getTopic(),msg.getTags(), new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
        try {
            returnJson = new String(msg.getBody(),"utf-8");
            logger.info("{},MQDistributedConsumer Thread handle message body: {}", Thread.currentThread().getName() , returnJson);
            if (StringUtils.isEmpty(returnJson)) {
                logger.error("===========>>>>>>CreditRocketMQ received message is null<<<<<<<===========");
                return;
            }
            // DistributedTaskExecutionContent 解析对应的实体，然后进行业务处理，处理完成之后发送消息
            DistributedTaskExecutionContent content = JSON.parseObject(returnJson, new TypeReference<DistributedTaskExecutionContent>() {}, Feature.SupportNonPublicField);
            DistributedTaskExecutionResult executionResult = getExecutionResult(content);

            callBackProducer.send(content.getMqCallback(),executionResult);
            logger.info("productId: {},totalNum: {},totalTime:{}   ------------------ rocketMQ调度消费执行成功" ,(new Date().getTime()-beginTime));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }

    public DistributedTaskExecutionResult getExecutionResult(DistributedTaskExecutionContent content){
        Integer partitionCount = content.getPartitionCount();
        Integer partitionIndex = content.getPartitionIndex();
        Map<String,Object> params = content.getParameters();// 如果生产者提供参数，取出之后进行后续的业务处理。
        DistributedTaskExecutionResult result = null;

        try{
            // 处理自己的业务逻辑
            if(params != null){
                logger.info("定时任务的消费端,总分区数{},当前执行的是第 {} 分区,参数params: {}",partitionCount,partitionIndex,params);
            }else {
                logger.info("定时任务的消费端,总分区数{},当前执行的是第 {} 分区", partitionCount, partitionIndex);
            }
            result = content.getSuccessResult();
            logger.info("定时任务消费成功");
        }catch (Exception e){
            result = content.getFailResult("消费消费过程中业务逻辑报错");
        }

        return result;
    }
}